package cn.tedu.luban.rocketmq.demo;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.List;

public class MyConsumer {

    @Test
    public void consumer() throws MQClientException {
        //1.连接RocketMQ
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr("localhost:9876");
        defaultMQPushConsumer.setConsumerGroup("consumer-group-01");
        //2.消费者所监听的目标信息，监听哪个主题,* 表示监听条件，主题中的所有信息
        defaultMQPushConsumer.subscribe("test-topic-01","*");
        //建立消费的监听，消息消费逻辑（打印）
        MessageListenerConcurrently messageListenerConcurrently=new MessageListenerConcurrently() {
            /**
             * 每次RoketMQ推送消息的时候，都会让监听器执行consumeMessage方法
             * @param list List<MessageExt> 表示推送过来的消息，可以处理消息的批量消费，当前不是批量消息，所以每个消息都会在list中存在唯一一个元素
             * @param consumeConcurrentlyContext 消费上下文，拿到消费者ID等信息
             * @return 反馈给RocketMQ消息消费状态
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //从list中解析出来需要的数据，body
                MessageExt messageExt = list.get(0);
                byte[] body = messageExt.getBody();//获取消息体
                String message = new String(body, StandardCharsets.UTF_8);
                System.out.println("消费者01 收到消息："+message);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        };

        defaultMQPushConsumer.setMessageListener(messageListenerConcurrently);
        //开启consumer的连接
        defaultMQPushConsumer.start();
        while (true);
    }
}
